package com.zhang.sql;

import com.zhang.gmall.realtime.utils.MyKafkaUtil;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.Table;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;

/**
 * @title:
 * @author: zhang
 * @date: 2022/3/15 17:37
 */
public class ProvinceStatsSqlApp {
    public static void main(String[] args) throws Exception {
        //TODO 1.获取流执行环境
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(4);
        //TODO 2.获取表执行环境
        StreamTableEnvironment tableEnv = StreamTableEnvironment.create(env);
        //TODO 3.检查点设置
        //TODO 4.从kafka读取数据创建动态表，生成watermark和提取事件时间字段
        String topic = "dwm_order_wide_2022";
        String groupID = "provinceStatsSqlApp";
        tableEnv.executeSql("create table order_wide (" +
                " order_id BIGINT," +
                " tm_id STRING ," +
                " province_id BIGINT," +
                " province_name STRING," +
                " province_area_code STRING," +
                " province_iso_code STRING," +
                " province_3166_2_code STRING," +
                " split_total_amount DOUBLE," +
                " create_time STRING ," +
                " proctime AS PROCTIME()"+
                ") WITH (" + MyKafkaUtil.getKafkaDDl(topic, groupID) +
                ")");


        tableEnv.executeSql("CREATE TABLE dim_base_trademark (" +
                " id STRING," +
                " INFO ROW<TM_NAME STRING>," +
                " PRIMARY KEY (id) NOT ENFORCED" +
                ") WITH (" +
                "'connector' = 'hbase-2.2'," +
                "'table-name' = 'GMALL_REALTIME_TEST:DIM_BASE_TRADEMARK'," +
                "'zookeeper.quorum' = 'hadoop102:2181'" +
                ")");

        Table table = tableEnv.sqlQuery(
                "SELECT tm_id,id,INFO FROM order_wide AS o LEFT JOIN dim_base_trademark FOR SYSTEM_TIME AS OF o.proctime AS u ON o.tm_id = u.id"
        );

        table.execute().print();
        //Table table = tableEnv.sqlQuery("select id,INFO.TM_NAME from dim_base_trademark");


        /*Table table = tableEnv.sqlQuery("select * from order_wide");
        table.execute().print();*/

        //TODO 5.从动态表中查询数据
        //TODO 1.
        //TODO 1.
        //TODO 1.
        //TODO 1.
        //TODO 1.
        //TODO 1.
        //TODO 1.
        //TODO 1.
        //TODO 1.
        //TODO 1.
        env.execute();
    }
}
